#include "CThreadPool.h"
#include "CMemory.h"
#include "CLogger.h"
#include "CLogicSocket.h"
#include <unistd.h>

pthread_mutex_t CThreadPool::m_pthreadMutex = PTHREAD_MUTEX_INITIALIZER;	//互斥量
pthread_cond_t CThreadPool::m_pthreadCond = PTHREAD_COND_INITIALIZER;	//条件变量
bool CThreadPool::m_bShutdown = false;				//线程退出标志

//构造函数
CThreadPool::CThreadPool()
{
	m_iRunningThreadNum = 0;	//正在运行的线程数目
	m_iLastEmgTime = 0;	//上次报告线程不够用的时间
	m_iMsgRecv = 0;		//收消息队列
}

CThreadPool::~CThreadPool()
{
	//接收消息队列中内容释放
	clearMsgRecvQueue();
}

//创建线程池中的线程，要手工调用，不在构造函数中调用了
//所有线程都创建成功，则返回true，出现错误返回false
bool CThreadPool::create(int nThreadNum)
{
	ThreadItem* pNew;
	int iErr;
	m_iThreadNum = nThreadNum;
	for (size_t i = 0; i < m_iThreadNum; i++)
	{
		m_vecThread.push_back(pNew = new ThreadItem(this));
		iErr = pthread_create(&pNew->m_tHandle, NULL, threadFunc, pNew);
		if (iErr != 0)
		{
			//创建线程出错
			CLogger::logStderr(iErr, "CThreadPool::create()创建线程%d失败，返回的错误码为%d！", i, iErr);
			return false;
		}
		else {
			//创建线程成功
			CLogger::logStderr(0, "CThreadPool::create()创建线程%d成功，线程id=%d", i, pNew->m_tHandle);
		}
	}

	//保证每个线程都启动并运行到pthread_cond_wait()，本函数才返回，只有这样，这几个线程才能进行后续的正常工作
lblFor:
	for (auto &it : m_vecThread)
	{
		if (it->m_bRunning == false)
		{
			//这说明有没有启动完成的线程
			usleep(100 * 1000);
			goto lblFor;
		}
	}

	return true;
}

//停止所有线程[等待结束线程池中所有线程，该函数返回后，所有线程池中线程都结束]
void CThreadPool::stopAll()
{
	//(1)已经调用过，就不要重复调用了
	if (m_bShutdown == true)
	{
		return;
	}
	m_bShutdown = true;

	//(2)唤醒等待该条件[卡在pthread_cond_wait()]的所有线程，一定要在改变条件变量状态后再给线程发信号
	int iErr = pthread_cond_broadcast(&m_pthreadCond);
	if (iErr != 0)
	{
		//打印紧急日志
		CLogger::logStderr(iErr, "CThreadPool::StopAll()中pthread_cond_broadcast()失败，返回的错误码为%d！", iErr);
		return;
	}
	//(3)等待线程，让线程真返回
	for (auto& it: m_vecThread)
	{
		pthread_join(it->m_tHandle, NULL);	//等待线程终止
	}

	//(4)流程走到这里，所有的线程池中的线程肯定都返回了
	pthread_mutex_destroy(&m_pthreadMutex);
	pthread_cond_destroy(&m_pthreadCond);

	//(5)释放new出来的ThreadItem[线程池中的线程]
	for (auto &it : m_vecThread)
	{
		if (it)
		{
			delete it;
		}
	}
	m_vecThread.clear();
	CLogger::logStderr(0, "CThreadPool::stopAll()成功返回，线程池中线程全部正常结束！");
}

//发送信号，并唤醒一个线程干活
void CThreadPool::call()
{
	int iErr = pthread_cond_signal(&m_pthreadCond);
	if (iErr != 0)
	{
		CLogger::logStderr(iErr, "CThreadPool::call()中pthread_cond_signal()失败，返回的错误码为%d！", iErr);
	}

	if (m_iThreadNum == m_iRunningThreadNum)	//线程池中线程总量，跟当前正在干活的线程数量一样，说明所有线程都忙碌起来，线程不够用了
	{
		//线程不够用了
		time_t currTime = time(NULL);
		if (currTime - m_iLastEmgTime > 10)	//最少间隔10秒报一次线程此中线程不够用的问题
		{
			m_iLastEmgTime = currTime;
			CLogger::logStderr(0, "CThreadPool::call()中发现线程池中当前空闲线程数量为0，要考虑扩容线程池了！");
		}
	}
}

//收到一个完整的消息后，入消息队列，并触发线程池中线程来处理该消息
void CThreadPool::inMsgRecvQueueAndSignal(char* pBuf)
{
	//互斥
	int iErr = pthread_mutex_lock(&m_pthreadMutex);
	if (iErr != 0)
	{
		CLogger::logStderr(iErr, "CThreadPool::inMsgRecvQueueAndSignal()中pthread_mutex_lock()失败，返回的错误码为%d！", iErr);
	}
	
	m_lstMsgRecv.push_back(pBuf);	//入消息队列
	++m_iMsgRecv;					//收消息队列条数+1，用变量比m_lstMsgRecv.size()更高效

	//取消互斥
	iErr = pthread_mutex_unlock(&m_pthreadMutex);
	if (iErr != 0)
	{
		CLogger::logStderr(iErr, "CThreadPool::inMsgRecvQueueAndSignal()中pthread_mutex_unlock失败，返回的错误码为%d！", iErr);
	}
	call();
}

//线程入口函数，当用pthread_create()创建线程后，这个threadFunc()函数都会被立即执行
void* CThreadPool::threadFunc(void* pParam)
{
	//这个是静态成员函数，是不存在this指针的
	ThreadItem* pThread = static_cast<ThreadItem*>(pParam);
	CThreadPool* pPool = pThread->m_pThis;

	CMemory* pMemory = CMemory::getInstance();
	int iErr;
	
	pthread_t tid = pthread_self();	//获取线程自身id，以方便调试打印信息等
	while (true)
	{
		iErr = pthread_mutex_lock(&m_pthreadMutex);
		if (iErr != 0)
		{
			CLogger::logStderr(iErr, "CThreadPool::threadFunc()中pthread_mutex_lock()失败，返回的错误码为%d!", iErr);
		}

		//pthread_cond_wait()是个值得注意的函数，调用一次pthread_cond_signal()可能会唤醒多个（惊群）
		//官方描述是至少一个，pthread_cond_signal在多处理器上可能同时唤醒多个线程
		while ((pPool->m_lstMsgRecv.size() == 0) && m_bShutdown == false) {
			if (pThread->m_bRunning == false)
			{
				pThread->m_bRunning = true;		//标记为true才允许调用StopAll()，线程执行到这里才认为是启动成功了
			}
			pthread_cond_wait(&m_pthreadCond, &m_pthreadMutex);	//整个服务器程序刚初始化的时候，所有线程必然是卡在这里等待的
		}

		//能走下来的，必然是拿到了真正的消息队列中的数据，或者m_bShutdown == true
		//先判断进程退出这个条件
		if (m_bShutdown)
		{
			pthread_mutex_unlock(&m_pthreadMutex);	//解锁互斥量
			break;
		}

		//走到这里，可以取到消息进行处理了
		char* pMsg = pPool->m_lstMsgRecv.front();	//返回第一个元素但不检查元素存在与否
		pPool->m_lstMsgRecv.pop_front();			//移除第一个元素但不返回
		--pPool->m_iMsgRecv;						//收消息队列中消息总数-1

		//可以解锁互斥量
		iErr = pthread_mutex_unlock(&m_pthreadMutex);
		if (iErr != 0)
		{
			//有问题及时报告
			CLogger::logStderr(iErr, "CThreadPool中ThreadFunc()中pthread_mutex_unlock()失败，返回的错误码为%d！", iErr);
		}

		//能走到这里说明有消息可以处理，开始处理
		++pPool->m_iRunningThreadNum;	//原子+1，记录正在干活的线程数量增加1，比互斥量要快很多

		//处理消息队列中的消息
		g_clLogicSocket.threadRecvProc(pMsg);

		pMemory->freeMemory(pMsg);		//释放消息内存
		--pPool->m_iRunningThreadNum;	//原子-1，记录正在干活的线程数量减少1
	}

	return nullptr;
}

void CThreadPool::clearMsgRecvQueue()
{
	char* pTmpMemPoint;
	CMemory* pMemory = CMemory::getInstance();

	while (!m_lstMsgRecv.empty())
	{
		pTmpMemPoint = m_lstMsgRecv.front();
		m_lstMsgRecv.pop_front();
		pMemory->freeMemory(pTmpMemPoint);
	}
}
